package utils

/**
  1. 可以设置工作线程
  2. 可以设置工作回调(无任务时会执行doIdle), 可以设置固定回调函数和休息间隔(优先使用),  二者只能回调一次
*/

import (
	"errors"
	"fmt"
	"os"
	"sync"
	"sync/atomic"
	"time"
)

const (
	max_queue_size     = 1024
	WORKER_STATE_END   = 0
	WORKER_STATE_ALIVE = 1
)

type WorkerRec struct {
	owner    *QueueTask
	idletime time.Duration
	state    byte
	TagPtr   interface{}
}

func (this *WorkerRec) State() byte {
	return this.state
}

func (this *WorkerRec) IdleTime() time.Duration {
	return this.idletime
}

func (this *WorkerRec) Owner() *QueueTask {
	return this.owner
}

type QueueTask struct {
	worker_num              int32
	min_work_cnt            int32
	max_work_cnt            int32
	push_fail_cnt           int32
	grow_worker_if_task_num int32 // Push任务时如果大于该值,则进行增长
	isTerminated            byte
	ontaskcb                queue_cb_event
	on_work_task_cb         func(worker *WorkerRec, args ...interface{})
	on_idle_cb              func(worker *WorkerRec)
	interval_idle_sleep     time.Duration
	onIntervalCb            func()
	dataQueue               *SyncQueue
	closeWg                 sync.WaitGroup
	max_queue_size          int
	Id                      string
	monitor_pop             *MonitorSpeedNTimeRec
	monitor_push            *MonitorSpeedNTimeRec
}

type queue_cb_event func(args ...interface{})

func NewQueueTaskFunc() *QueueTask {
	rval := &QueueTask{
		dataQueue:               NewSyncQueue(),
		max_queue_size:          max_queue_size,
		grow_worker_if_task_num: 1,
		interval_idle_sleep:     time.Millisecond * 100,
	}
	return rval
}

func NewQueueTask(ontaskcb queue_cb_event) *QueueTask {
	rval := &QueueTask{
		ontaskcb:                ontaskcb,
		dataQueue:               NewSyncQueue(),
		max_queue_size:          max_queue_size,
		interval_idle_sleep:     time.Millisecond * 100,
		grow_worker_if_task_num: 1,
	}
	return rval
}

func NewQueueTaskWithMaxSize(ontaskcb queue_cb_event, max_size int) *QueueTask {
	rval := &QueueTask{
		ontaskcb:                ontaskcb,
		dataQueue:               NewSyncQueue(),
		max_queue_size:          max_size,
		interval_idle_sleep:     time.Millisecond * 100,
		grow_worker_if_task_num: 1,
	}
	return rval
}

func (this *QueueTask) ConfigMonitor(pop, push *MonitorSpeedNTimeRec) {
	this.monitor_pop = pop
	this.monitor_push = push
}

func (this *QueueTask) ConfigTaskCallBackFunc(ontaskcb queue_cb_event) *QueueTask {
	this.ontaskcb = ontaskcb
	return this
}

func (this *QueueTask) ConfigWorkCallBackCallBackFunc(cb func(worker *WorkerRec, args ...interface{})) *QueueTask {
	this.on_work_task_cb = cb
	return this
}

func (this *QueueTask) ConfigIdleCallBackFunc(onIdleCb func(worker *WorkerRec)) *QueueTask {
	this.on_idle_cb = onIdleCb
	return this
}

/*
**

	idle_sleep_interval 如果任务为0时, 进行休息时间, 如果任务不为0, 则不进行休息
*/
func (this *QueueTask) ConfigIntervalCallBack(idle_sleep_interval time.Duration, cb func()) *QueueTask {
	this.ConfigIdleSleepTime(idle_sleep_interval)
	this.onIntervalCb = cb
	return this
}

func (this *QueueTask) ConfigIdleSleepTime(val time.Duration) *QueueTask {
	if this.interval_idle_sleep == 0 {
		this.interval_idle_sleep = time.Millisecond * 100
	} else {
		this.interval_idle_sleep = val
	}
	return this
}

func (this *QueueTask) ConfigMaxQueueSize(val int) *QueueTask {
	this.max_queue_size = val
	return this
}

func (this *QueueTask) ConfigWorkNum(min, max int32) *QueueTask {

	this.max_work_cnt = max
	this.min_work_cnt = min

	return this
}

func (this *QueueTask) ConfigMaxNum(max int32) *QueueTask {
	this.max_work_cnt = max
	return this
}

func (this *QueueTask) checkGrowWork() {
	r := atomic.AddInt32(&this.worker_num, 1)
	if r > this.max_work_cnt {
		atomic.AddInt32(&this.worker_num, -1)
		return
	}
	GoFunc2(this.doWorker)
}

func (this *QueueTask) doIdle(worker *WorkerRec) {
	if this.on_idle_cb != nil {
		this.on_idle_cb(worker)
	}
}

func (this *QueueTask) StatusString() string {
	var jb BytesBuilder
	jb.Appendf("max_size:%d,remain:%d,worknum:%d/%d", this.max_queue_size, this.dataQueue.Count(), this.worker_num, this.max_work_cnt)
	if this.push_fail_cnt > 0 {
		jb.Appendf(",push_fail:%d", this.push_fail_cnt)
	}

	mon := this.monitor_pop
	if mon != nil {
		jb.Appendf("pop:%s", mon.Info())
	}
	{
		mon := this.monitor_push
		if mon != nil {
			jb.Appendf("push:%s", mon.Info())
		}
	}
	return jb.String()
}

func (this *QueueTask) String() string {
	var jb JSONBuilder
	jb.ObjectBegin()
	jb.AddValues("max_size", this.max_queue_size, "remain", this.dataQueue.Count(), "worknum", this.worker_num)
	if this.push_fail_cnt > 0 {
		jb.Spliter()
		jb.Add("push_fail_cnt", this.push_fail_cnt)
	}

	mon := this.monitor_pop
	if mon != nil {
		jb.Spliter()
		jb.AddValues("pop", mon.SpeedIO())
	}
	{
		mon := this.monitor_push
		if mon != nil {
			jb.Spliter()
			jb.AddValues("push", mon.SpeedIO())
		}
	}

	jb.ObjectEnd()
	return jb.String()
}

/**
 * 取一个任务
 */
func (this *QueueTask) PopTaskArg() []interface{} {
	if ok, val := this.dataQueue.Pop(); ok {
		mon := this.monitor_pop
		if mon != nil {
			mon.Delta(1)
		}
		return val.([]interface{})
	} else {
		return nil
	}
}

/**
 * 取多个任务, 一般在回调中执行
 */
func (this *QueueTask) PopTaskArgMaxFunc(max int, cb func(args ...interface{}) bool) int {
	i := 0
	for i < max {
		if ok, val := this.dataQueue.Pop(); ok {
			mon := this.monitor_pop
			if mon != nil {
				mon.Delta(1)
			}
			vals := val.([]interface{})
			i++
			if !cb(vals...) {
				break
			}
		} else {
			break
		}
	}
	return i
}

func (this *QueueTask) doWorker() {
	this.closeWg.Add(1)

	defer func() {
		this.closeWg.Done()
		atomic.AddInt32(&this.worker_num, -1)
	}()

	worker := &WorkerRec{owner: this, idletime: 0, state: WORKER_STATE_ALIVE}

	for {
		if this.isTerminated == 1 {
			break
		}

		if this.onIntervalCb != nil {
			this.onIntervalCb()
			if this.dataQueue.Count() == 0 {
				if worker.idletime > time.Second {
					if this.worker_num > this.min_work_cnt {
						break // 1秒没有任务, 进行释放worker
					}
				}
				this.doIdle(worker)
				time.Sleep(this.interval_idle_sleep)
				worker.idletime += this.interval_idle_sleep
			}
			continue
		} else if this.on_work_task_cb != nil {
			if ok, val := this.dataQueue.Pop(); ok {
				mon := this.monitor_pop
				if mon != nil {
					mon.Delta(1)
				}
				worker.idletime = 0
				if args, isOK := val.([]interface{}); isOK {
					this.on_work_task_cb(worker, args...)
				} else {
					this.on_work_task_cb(worker, args)
				}
			} else {
				if worker.idletime > time.Second {
					if this.worker_num > this.min_work_cnt {
						break // 1秒没有任务, 进行释放worker
					}
				}
				this.doIdle(worker)
				time.Sleep(this.interval_idle_sleep)
				worker.idletime += this.interval_idle_sleep
				continue // 没有需要工作的
			}
		} else if this.ontaskcb != nil {
			if ok, val := this.dataQueue.Pop(); ok {
				mon := this.monitor_pop
				if mon != nil {
					mon.Delta(1)
				}
				worker.idletime = 0
				if args, isOK := val.([]interface{}); isOK {
					this.ontaskcb(args...)
				} else {
					this.ontaskcb(args)
				}
			} else {
				if worker.idletime > time.Second {
					if this.worker_num > this.min_work_cnt {
						break // 1秒没有任务, 进行释放worker
					}
				}
				this.doIdle(worker)
				time.Sleep(this.interval_idle_sleep)
				worker.idletime += this.interval_idle_sleep
				continue // 没有需要工作的
			}
		} else {
			if worker.idletime > time.Second {
				if this.worker_num > this.min_work_cnt {
					break // 1秒没有任务, 进行释放worker
				}
			}
			this.doIdle(worker)
			time.Sleep(this.interval_idle_sleep)
			worker.idletime += this.interval_idle_sleep
			continue // 没有需要工作的
		}
	}

	worker.state = WORKER_STATE_END
	this.doIdle(worker)
}

func (this *QueueTask) Start() *QueueTask {
	this.isTerminated = 0
	if this.ontaskcb == nil && this.onIntervalCb == nil && this.on_work_task_cb == nil {
		fmt.Fprintf(os.Stderr, "[QueueTask.Start()]回调事件没有赋值！")
	}
	if this.min_work_cnt <= 0 {
		this.min_work_cnt = 1
	}
	if this.max_work_cnt <= 0 {
		this.max_work_cnt = 1
	}

	for this.worker_num < this.min_work_cnt {
		this.checkGrowWork()
	}

	return this
}

func (this *QueueTask) Stop() {
	this.isTerminated = 1
	this.closeWg.Wait()
}

/*
**

	注意，入队列的数据，最好是Copy一份, 避免异步修改, 或者不要进行异步修改
*/
func (this *QueueTask) Push(args ...interface{}) error {
	if this.isTerminated == 1 {
		return errors.New("任务已经请求停止！")

	}
	if int(this.dataQueue.Count()) >= this.max_queue_size {
		atomic.AddInt32(&this.push_fail_cnt, 1)
		return errors.New(fmt.Sprintf("队列超过堆积数量[%d]！", this.max_queue_size))
	}

	this.dataQueue.Push(args)

	mon := this.monitor_push
	if mon != nil {
		mon.Delta(1)
	}

	if this.grow_worker_if_task_num > 0 && this.dataQueue.Count() > this.grow_worker_if_task_num { // 大于100个任务未执行
		this.checkGrowWork() // 增长worker
	}
	return nil
}

func (this *QueueTask) Count() int32 {
	return this.dataQueue.Count()
}
